import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.BatchTableEnvironment;
import org.apache.flink.types.Row;

import java.util.Arrays;
import static org.apache.flink.table.api.Expressions.$;


public class OrderBy
{

    public static void main(String[] args) throws Exception
    {


        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        BatchTableEnvironment tEnv = BatchTableEnvironment.create(env);

        DataSet<Order> ds = env.fromCollection(Arrays.asList(
                new Order(1L, "beer", 3),
                new Order(3L, "rubber", 2),
                new Order(1L, "diaper", 4)
        ));

        Table in = tEnv.fromDataSet(ds, "user,product,amount");
        Table result = in.orderBy($("amount").asc());

        tEnv.toDataSet(result, Row.class).print();
}
}
